package gr.unipi.rtopk;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class RtopK {
	public static class Tuple {
		private int id;
		private double[] values;
		
		public Tuple(int id, double[] values) {
			this.id = id;
			this.values = values;
		}
		
		public Tuple(String text) {
			String[] tokens = text.split(" ");
			
			id = Integer.parseInt(tokens[0]);
			
			values = new double[tokens.length - 1];
			for (int i = 1; i < tokens.length; i++) {
				values[i - 1] = Double.parseDouble(tokens[i]);
			}
		}

		public int getId() {
			return id;
		}

		public void setId(int id) {
			this.id = id;
		}

		public double[] getValues() {
			return values;
		}

		public void setValues(double[] values) {
			this.values = values;
		}
	}
	
	public static class ScoreMapper extends Mapper<Object, Text, IntWritable, IntWritable> {
		
		private FileSystem fileSystem;
		private Path pt;
		private Tuple qTuple;
		private IntWritable one = new IntWritable(1);
		
		private double calculatefunction(Tuple p, Tuple w) throws IOException {
			if (p.getValues().length != w.getValues().length) {
				throw new IOException("Different dimensions detected in p and w.");
			}
			
			double sum = 0;
			for (int i = 0; i < p.getValues().length; i++) {
				sum += w.getValues()[i] * p.getValues()[i];
			}
			
			return sum;
		}
		
		@Override
		protected void setup(Context context) throws IOException {
			String q = context.getConfiguration().get("q");
			qTuple = new Tuple(q);
			
			pt = new Path("W.txt");
			fileSystem =  FileSystem.getLocal(context.getConfiguration()).getRawFileSystem();
			//FSDataInputStream inputStream = ;
			//inputStream.
			//br = new BufferedReader(new InputStreamReader(fileSystem.open(pt)));
		}
		
		@Override
		protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
			Tuple p = new Tuple(value.toString());

			BufferedReader br = new BufferedReader(new InputStreamReader(fileSystem.open(pt)));
			try {
				String line;
				line = br.readLine();
				Tuple w;
				while (line != null) {
					w = new Tuple(line);
					double scoreP = calculatefunction(p, w);
					double scoreQ = calculatefunction(qTuple, w);
					if (scoreP < scoreQ) {
						context.write(new IntWritable(w.getId()), one);
					}

					// be sure to read the next line otherwise you'll get an
					// infinite loop
					line = br.readLine();
				}
			} finally {
				// you should close out the BufferedReader
				br.close();
			}
		}
	}
	
	public static class RtopkCombiner extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable> {
		public void reduce(IntWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
			int sum = 0;
			for (IntWritable value: values) {
				sum += value.get();
			}
			IntWritable result = new IntWritable(sum);
			context.write(key, result);
		}
	}

	public static class RtopkReducer extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable> {
		private int k;
		
		@Override
		protected void setup(Context context) throws IOException {
			String kString = context.getConfiguration().get("k");
			k = Integer.parseInt(kString);
		}
		
		public void reduce(IntWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
			int sum = 0;
			for (IntWritable value: values) {
				sum += value.get();
			}
			if (sum < k) {
				IntWritable result = new IntWritable(sum);
				context.write(key, result);
			}
		}
	}

	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		Path pt = new Path("input/Q.txt");
		FileSystem fs = FileSystem.get(conf);
		BufferedReader br = new BufferedReader(new InputStreamReader(fs.open(pt)));
		try {
			String line;
			line = br.readLine();
			if (line != null) {
				//System.out.println("q = " + line);
				conf.set("q", line);
			}
		} finally {
			// you should close out the BufferedReader
			br.close();
		}
		conf.set("k", args[0]);
		Job job = Job.getInstance(conf, "reverse topk");
		job.setJarByClass(RtopK.class);
		job.setMapperClass(ScoreMapper.class);
		job.setCombinerClass(RtopkCombiner.class);
		job.setReducerClass(RtopkReducer.class);
		job.setOutputKeyClass(IntWritable.class);
		job.setOutputValueClass(IntWritable.class);
		FileInputFormat.addInputPath(job, new Path("input/S.txt"));
		FileOutputFormat.setOutputPath(job, new Path("output"));
		System.exit(job.waitForCompletion(true) ? 0 : 1);
	}
}
